package br.pucrs.lognplay.job;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.List;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.log4j.Logger;
import org.elasticsearch.common.joda.time.DateTime;
import org.elasticsearch.common.joda.time.format.DateTimeFormat;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;

import br.pucrs.lognplay.entity.LogSource;
import br.pucrs.lognplay.entity.StaticView;
import br.pucrs.lognplay.job.mapper.IndexerJobMapper;
import br.pucrs.lognplay.repository.RepositoryException;
import br.pucrs.lognplay.repository.RepositoryManager;
import br.pucrs.lognplay.repository.StaticViewRepository;

public class StaticIndexerJob implements Job {

	private static final Logger logger = Logger.getLogger(StaticIndexerJob.class);

	public static final String STATIC_VIEW_ID_PARAMETER = "STATIC_VIEW_ID";

	private static final String HDFS_TMP_DIR = "/tmp/";

	private static final String INDEXER_JOB_MAPPER_NAME = "";

	private StaticView staticView = null;

	public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
		try {
			logger.trace("================================ StaticIndexerJob : START ===================================");
			String staticViewId = jobExecutionContext.getJobDetail().getJobDataMap().getString(STATIC_VIEW_ID_PARAMETER);

			RepositoryManager.beginTransaction();
			StaticViewRepository staticViewRepository = RepositoryManager.getStaticViewRepository();
			staticView = staticViewRepository.get(staticViewId);
			RepositoryManager.commitTransaction();

			logger.trace(staticView.getStatus());

			if (staticView.getStatus() == StaticView.Status.PENDING_INDEXING) {
				processPendingIndexing();
			} else if (staticView.getStatus() == StaticView.Status.PENDING_REMOVAL) {
				processPendingRemoval();
			}

			logger.trace("=============================== StaticIndexerJob : FINISH ===================================");
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private void processPendingIndexing() {
		logger.trace("processPendingIndexing");
		for (LogSource logSource : staticView.getLogSources()) {
			processLogSource(logSource);
		}

		/* Update Status */
		try {
			logger.trace("Updating static view status");
			staticView.setStatus(StaticView.Status.INDEXED);
			RepositoryManager.beginTransaction();
			StaticViewRepository staticViewRepository = RepositoryManager.getStaticViewRepository();
			staticViewRepository.merge(staticView); // TODO
			RepositoryManager.commitTransaction();
		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	private void processLogSource(LogSource logSource) {
		logger.trace("processLogSource: " + logSource.getName());

		try {

			List<String> filesToProcess = new ArrayList<String>();

			String clientId = logSource.getClient().getId().toLowerCase().replaceAll("-", "");
			String logSourceId = logSource.getId().toLowerCase().replaceAll("-", "");

			String clientPath = "/usr/lognplay/" + clientId + Path.SEPARATOR_CHAR;
			String sourcePath = clientPath + logSourceId + Path.SEPARATOR_CHAR;
			String filePath;

			Calendar fromDate = Calendar.getInstance();
			Calendar toDate = Calendar.getInstance();
			Calendar endDate = Calendar.getInstance();

			fromDate.setTime(staticView.getFrom());
			toDate.setTime(staticView.getFrom());
			endDate.setTime(staticView.getTo());

			toDate.add(Calendar.HOUR, 1);

			while (fromDate.before(endDate)) {
				filePath = sourcePath + formatFileName(fromDate, toDate);
				if (HDFSTools.getFileSystem().exists(new Path(filePath))) {
					filesToProcess.add(filePath);
				}
				fromDate.add(Calendar.HOUR, 1);
				toDate.add(Calendar.HOUR, 1);
			}

			if (filesToProcess.size() > 0) {
				setupIndexerJob(filesToProcess);
			} else {
				logger.trace("No files!");
			}

		} catch (Exception e) {
			logger.error(e);
		}

	}

	private void setupIndexerJob(List<String> filesToProcess) {
		logger.trace("======== setupIndexerJob ========");
		logger.trace("staticViewId: " + staticView.getId());
		logger.trace("filesToProcess: " + filesToProcess);
		try {
			String outputPath = HDFS_TMP_DIR + System.currentTimeMillis();

			JobConf conf = new JobConf(HDFSTools.getConfiguration(), StaticIndexerJob.class);

			conf.setSpeculativeExecution(false);
			conf.setJobName(INDEXER_JOB_MAPPER_NAME + "#" + staticView.getId());
			conf.setInputFormat(TextInputFormat.class);
			conf.setMapperClass(IndexerJobMapper.class);
			conf.setMapOutputKeyClass(NullWritable.class);
			conf.setMapOutputValueClass(NullWritable.class);
			conf.setNumReduceTasks(0);

			conf.set(IndexerJobMapper.STATIC_VIEW_ID_PARAMETER, staticView.getId().toLowerCase().replaceAll("-", ""));
			conf.set(IndexerJobMapper.OUTPUT_PATH_PARAMETER, outputPath);

			for (String filePath : filesToProcess) {
				FileInputFormat.addInputPath(conf, new Path(filePath));
			}

			FileOutputFormat.setOutputPath(conf, new Path(outputPath));

			JobClient.runJob(conf);
		} catch (IOException e) {
			logger.error(e);
		}
	}

	private void processPendingRemoval() {
		logger.trace("processPendingRemoval");

		try {
			ElasticSearchClient esClient = new ElasticSearchClient();
			String staticViewId = staticView.getId().toLowerCase().replace("-", "");

			logger.trace("Removing Index: " + staticViewId);

			esClient.startClient();
			esClient.getEsClient().prepareDelete().setIndex(staticViewId).setOperationThreaded(false).execute().actionGet();
			esClient.stopClient();

			/* Update Status */

			staticView.setStatus(StaticView.Status.REMOVED);

			RepositoryManager.beginTransaction();
			StaticViewRepository staticViewRepository = RepositoryManager.getStaticViewRepository();
			staticViewRepository.merge(staticView);
			RepositoryManager.commitTransaction();

		} catch (RepositoryException e) {
			logger.error(e);
		}

	}

	private String formatDate(Calendar date, String pattern) {
		return DateTimeFormat.forPattern(pattern).print(new DateTime(date));
	}

	private String formatFileName(Calendar fromDate, Calendar toDate) {
		String fileName = formatDate(fromDate, "YYYY-MM-dd_HHmm-");
		fileName += formatDate(toDate, "HHmm");
		fileName += ".log";
		return fileName;
	}

}
